package com.bw.gmall.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.common.base.BaseApp;
import com.bw.gmall.realtime.common.bean.TradeOrderBean;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.function.DorisMapFunction;
import com.bw.gmall.realtime.common.util.DateFormatUtil;
import com.bw.gmall.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.hadoop.crypto.key.JavaKeyStoreProvider;

import java.time.Duration;
import java.util.Iterator;

public class DwsTradeOrderWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsTradeOrderWindow().start(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL,Constant.DWS_TRADE_ORDER_WINDOW,4,10028);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> dataStreamSource) {

        //ELT
//        dataStreamSource.print();
        SingleOutputStreamOperator<JSONObject> streamOperator = EltStream(dataStreamSource);

        //水位线

        SingleOutputStreamOperator<JSONObject> waterStream = waterStream(streamOperator);


        // 分组 聚合 求  下单独立用户数  和 下单新用户数
        SingleOutputStreamOperator<TradeOrderBean> uVnewCVStream = KeyUVnewCVStream(waterStream);
        uVnewCVStream.print();

        // 开窗 reduce

        SingleOutputStreamOperator<TradeOrderBean> streamOperator1 = WindowStream(uVnewCVStream);

        //写入doris
        streamOperator1.map(new DorisMapFunction<>()).sinkTo(FlinkSinkUtil.getDorisSinks(Constant.DWS_TRADE_ORDER_WINDOW));


    }

    private static SingleOutputStreamOperator<TradeOrderBean> WindowStream(SingleOutputStreamOperator<TradeOrderBean> uVnewCVStream) {
         return uVnewCVStream.windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<TradeOrderBean>() {
                    @Override
                    public TradeOrderBean reduce(TradeOrderBean t1, TradeOrderBean t2) throws Exception {
                        t1.setOrderNewUserCount(t1.getOrderNewUserCount() + t2.getOrderNewUserCount());
                        t1.setOrderUniqueUserCount(t1.getOrderUniqueUserCount() + t2.getOrderUniqueUserCount());
                        return t1;
                    }
                }, new AllWindowFunction<TradeOrderBean, TradeOrderBean, TimeWindow>() {
                    @Override
                    public void apply(TimeWindow timeWindow, Iterable<TradeOrderBean> iterable, Collector<TradeOrderBean> collector) throws Exception {

                        long start = timeWindow.getStart();
                        long end = timeWindow.getEnd();

                        Iterator<TradeOrderBean> iterator = iterable.iterator();
                        TradeOrderBean tradeOrderBean = iterator.next();

                        tradeOrderBean.setStt(DateFormatUtil.tsToDateTime(start));
                        tradeOrderBean.setEdt(DateFormatUtil.tsToDateTime(end));
                        tradeOrderBean.setCurDate(DateFormatUtil.tsToDateTime(System.currentTimeMillis()));
                        collector.collect(tradeOrderBean);
                    }
                });
    }

    private static SingleOutputStreamOperator<TradeOrderBean> KeyUVnewCVStream(SingleOutputStreamOperator<JSONObject> waterStream) {
         return  waterStream.keyBy(x -> x.getString("user_id")).process(new ProcessFunction<JSONObject, TradeOrderBean>() {
            ValueState<String> state;


            @Override
            public void open(Configuration parameters) {
                ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("state", String.class);
                state = getRuntimeContext().getState(stateDescriptor);
            }

            @Override
            public void processElement(JSONObject jsonObject, ProcessFunction<JSONObject, TradeOrderBean>.Context context, Collector<TradeOrderBean> collector) throws Exception {
                Long ts = jsonObject.getLong("ts");

                String tsToDate = DateFormatUtil.tsToDateTime(ts);

                String value = state.value();
                long uv = 0L;
                long newuv = 0L;

                if (!tsToDate.equals(value)) {
                    uv = 1L;
                    state.update(tsToDate);

                    if (value == null) {
                        newuv = 1L;
                    }
                }

                if (uv == 1) {
                    collector.collect(TradeOrderBean.builder()
                            .orderUniqueUserCount(uv)
                            .orderNewUserCount(newuv)
                            .ts(ts)
                            .curDate(tsToDate)
                            .build());
                }
            }
        });
    }

    private static SingleOutputStreamOperator<JSONObject> waterStream(SingleOutputStreamOperator<JSONObject> streamOperator) {
         return streamOperator.assignTimestampsAndWatermarks(WatermarkStrategy
                .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                .withTimestampAssigner((event, timestamp) -> event.getLong("ts")).withIdleness(Duration.ofSeconds(1)));
    }

    private static SingleOutputStreamOperator<JSONObject> EltStream(DataStreamSource<String> dataStreamSource) {
         return dataStreamSource.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String s, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                if (s != null) {

                    JSONObject jsonObject = JSON.parseObject(s);
                    if (jsonObject != null) {

                        Long ts = jsonObject.getLong("ts");
                        String user_id = jsonObject.getString("user_id");

                        if (ts != null && user_id != null) {
                            jsonObject.put("ts",ts*1000);

                            collector.collect(jsonObject);
                        }
                    }
                }
            }
        });
    }
}
